package bupt.xixi.registry.impl;

import lombok.extern.slf4j.Slf4j;
import org.apache.curator.RetryPolicy;
import org.apache.curator.framework.CuratorFramework;
import org.apache.curator.framework.CuratorFrameworkFactory;
import org.apache.curator.framework.recipes.cache.PathChildrenCache;
import org.apache.curator.framework.recipes.cache.PathChildrenCacheListener;
import org.apache.curator.retry.ExponentialBackoffRetry;

import java.util.List;

@Slf4j
public class ZkServiceDiscovery extends AbstractDiscovery {

    private static final String ZK_REGISTER_ROOT_PATH = "/my-rpc";
    private CuratorFramework zkClient;

    @Override
    protected List<String> doLookupService(String serviceName) {
        if (zkClient == null) {
            connect();
        }
        // 2. 如果没有, 去ZK找该节点子节点
        List<String> result = null;
        String servicePath = ZK_REGISTER_ROOT_PATH + "/" + serviceName;
        try {
            // 2.1 获取servicePath下所有子节点
            result = zkClient.getChildren().forPath(servicePath);
            // 2.3 注册监听器, 监视这个service下的改变, 可能会增加/减少提供服务的节点
            if (!CACHE_SERVICE_ADDRESS_MAP.containsKey(serviceName)) {
                registerWatcher(serviceName);
                CACHE_SERVICE_ADDRESS_MAP.put(serviceName,
                                              zkClient.getChildren().forPath(ZK_REGISTER_ROOT_PATH + "/" + serviceName));
            }
        } catch (Exception e) {
            log.error("查找远程方法失败 [{}]", servicePath);
        }
        return result;
    }

    private void registerWatcher(String rpcServiceName) throws Exception {
        // 得到该父节点地址
        String servicePath = ZK_REGISTER_ROOT_PATH + "/" + rpcServiceName;
        // 开辟一块缓存记录该父节点下子节点的数据
        PathChildrenCache pathChildrenCache = new PathChildrenCache(zkClient, servicePath, true);
        // 创建监听器监测缓存变化: 通过lambda表达式实现单函数接口, 设置回调函数
        PathChildrenCacheListener pathChildrenCacheListener = (curatorFramework, pathChildrenCacheEvent) -> {
            // 只要子节点发生变化: 增删改等
            // 就更新内存中的服务列表!
            List<String> serviceAddresses = curatorFramework.getChildren().forPath(servicePath);
            CACHE_SERVICE_ADDRESS_MAP.put(rpcServiceName, serviceAddresses);

        };
        // 注册监听器
        pathChildrenCache.getListenable().addListener(pathChildrenCacheListener);
        // 开启监听器
        pathChildrenCache.start();
    }

    private void connect() {
        // 设置默认重连策略, 重连三次&每次间隔1s
        RetryPolicy retryPolicy = new ExponentialBackoffRetry(BASE_SLEEP_TIME, MAX_RETRIES);
        zkClient = CuratorFrameworkFactory.builder().connectString("127.0.0.1:2181").retryPolicy(retryPolicy).build();
        zkClient.start();
    }
}
